/*
 * threadsPPD.c
 *
 *  Created on: 21/11/2011
 *      Author: utn_so
 */

#include "../../headers/GlobalHeader.h"
extern int32_t contador_msync, cantidadClientesPFS;
extern int32_t filesize;
extern pthread_mutex_t mutexListaPedidos, mutexColaSSTF;
extern Nodo *lista_pedidos;
extern Nodo *fin_lista_pedidos;
extern NodoNstep *fin_lista_pedidosnstep;
extern char planificador[30];
extern NodoNstep *lista_pedidosnstep;
extern sem_t *sem_lista, *enviePedido, *respuesta;
extern struct_config configuracion;
extern PPD_CHS posicionActualCabezal, posicionAnterior;
extern int32_t cantidadTotalPedidos;
extern t_log *logger;
extern int32_t modoInicio, tipoPedido, algoritmoUtilizado;
extern t_Cola *colaPedidos;
extern int32_t sizeCola, sizeListaNSTEP;
extern NodoPFS *listaPFS;
int32_t sd = -1;

void thread_cliente_pfs (void *parametro)
{
	int32_t addrlen = sizeof(struct sockaddr_in);
	int32_t socket_cliente = socket(AF_INET,SOCK_STREAM,0);
	int32_t rc;
	char carac;
	nipc paquete_mando,paquete_recibo;
	int32_t hand_ok;

	//Direccion local pfs
	struct sockaddr_in *direccion_local = malloc(addrlen);
	//Direccion remota(a la que me quiero conectar)
	struct sockaddr_in *direccion_remota = malloc(addrlen);

	direccion_local->sin_family = AF_INET;
	direccion_local->sin_addr.s_addr = INADDR_ANY;
	direccion_local->sin_port = htons(configuracion.PuertoEscuchaPpd);
	rc = bind(socket_cliente,(struct sockaddr *)direccion_local, addrlen);
	if (rc < 0)
	{
		perror("bind() socket_cliente");
	}

	direccion_remota->sin_family = AF_INET;
	direccion_remota->sin_addr.s_addr = inet_addr(configuracion.IPRaid);
	direccion_remota->sin_port = htons(configuracion.PuertoRaid);

	connect(socket_cliente,(struct sockaddr *)direccion_remota,addrlen);

	paquete_mando.type = 0;
	paquete_mando.payloadlength = 0;
	if((send(socket_cliente,&paquete_mando,sizeof(nipc),0)) <= 0){
		perror("send");
	}

	if((recv(socket_cliente,&paquete_recibo,sizeof(nipc),0)) <= 0) {
		perror("recv");
		exit (1);
	}

	if(paquete_recibo.payloadlength == 0){
		paquete_mando.type = 1;
		paquete_mando.payload.direccionSector = 555;
		paquete_mando.payload.conexion_pfs_id = socket_cliente;
		paquete_mando.payload.lectura_o_escritura = 'L';
		if((send(socket_cliente,&paquete_mando,sizeof(nipc),0)) <= 0){
			perror("send");
		}

		if((recv(socket_cliente,&paquete_recibo,sizeof(nipc),0)) <= 0) {
			perror("recv");
			exit (1);
		}
		if(paquete_recibo.type == 11){
			//		printf("informacion sector: %s",paquete_recibo.payload.info);
		}
	}
}


void insertarEnArray(int32_t array[CANT_CLIENTES_PFS_MAX],int32_t socket_id){
	int32_t i = 0;

	while(array[i] != -1){
		i++;
	}
	array[i] = socket_id;
}

void inicializoArray(int32_t array[CANT_CLIENTES_PFS_MAX]){
	int32_t i;
	for(i=0 ; i < CANT_CLIENTES_PFS_MAX; i++){
		array[i] = -1;
	}
}
void eliminoDeArray(int32_t array[CANT_CLIENTES_PFS_MAX],int32_t socket_id){
	int32_t i = 0 ,o = 0;
	while(array[i] != socket_id){
		i++;
	}
	for (o = i; i == CANT_CLIENTES_PFS_MAX; o++){
		array[o] = array[o+1];
	}
}

void thread_msync (){
	int32_t valorRetorno;
	// tim.tv_sec = 1;
	typedef struct timespec{
		time_t tv_sec;
		long tv_nsec;
	}timespec_t;
	struct timespec tim, tim2;
	tim.tv_nsec = 500000000;

	while(1){
		if(contador_msync > 30){
			if((valorRetorno=msync(handlerArchivoMapeado, filesize*1024, MS_SYNC)) == -1 ){
				perror("msync erroneo1");
				cerrarArchivo();
				exit(1);
			}
			contador_msync = 0;
		}else{
			nanosleep(&tim,NULL);
			if((valorRetorno=msync(handlerArchivoMapeado, filesize*1024, MS_SYNC)) == -1 ){
				perror("msync erroneo2");
				cerrarArchivo();
				exit(1);
			}
		}
	}
}

void thread_lector_pedidos_raid (void *parametro)
{
	struct_thread_lector struct_thread = *(struct_thread_lector *) parametro;
	int32_t socket_id = struct_thread.socket_id;
	int32_t rc;
	nipc paquete;

	while(1)
	{
		rc = recv(socket_id,&paquete,sizeof(nipc),MSG_WAITALL);
		if (rc < 0)
		{
			if (configuracion.LogActivado==1)
				log_error(logger, "THREAD_LECTOR_PEDIDOS_RAID", "Error al recibir pedido de RAID");
			cerrarArchivo();
			exit(1);
		}

		PPD_CHS estructura;
		estructura = ppd_LBAToCHS(paquete.payload.direccionSector);
		if (configuracion.LogActivado==1)
			log_info(logger, "MAIN", "Agregue pedido nro: %i:%i\n",estructura.cylinder, estructura.sector);

		//pthread_mutex_lock (&mutexListaPedidos);

		//printf("agregue %i ;;", paquete.payload.direccionSector);
		pthread_mutex_lock (&mutexListaPedidos);
		agregarPedido_cola(paquete.payload);
		pthread_mutex_unlock (&mutexListaPedidos);

		//pthread_mutex_unlock (&mutexListaPedidos);


	}
}

void thread_lector_consola (void *parametro)
{
	struct_thread_lector struct_thread_consola = *(struct_thread_lector *) parametro;
	int resultado;
	cabecera_paquete cabeceraRecibo;
	protocoloINFO_respuesta *respuestaINFO;
	protocoloCLEAN_respuesta *respuestaCLEAN;
	protocoloTRACE_respuesta *respuestaTRACE;
	pedido_clean pedidoClean;
	pedido_t1 pedidoTrace1;
	pedido_t2 pedidoTrace2;
	pedido_t3 pedidoTrace3;
	pedido_t4 pedidoTrace4;
	pedido_t5 pedidoTrace5;
	t_stream *streamEnvio;
	t_pedido pedidoAPlanificar;
	int32_t rc, contador, cantidadSectores, current, i;
	int32_t *sectores, *sectoresAPlanificar;
	struct sockaddr_un serveraddr;

	char sectoresPorPista[10];
	char cabezalesPorCilindro[2];
	char cilindros[10];
	sprintf(sectoresPorPista,"%d",configuracion.Sectors);
	sprintf(cabezalesPorCilindro, "%d", configuracion.Heads);
	sprintf(cilindros, "%d", configuracion.Cylinders);

	const char* args[] = {configuracion.SocketUNIX,cabezalesPorCilindro,sectoresPorPista,cilindros, NULL};

	posicionActualCabezal=ppd_LBAToCHS(0);

	PPD_CHS sectorPlanifico, sectorAnterior, sectorInicial, sectorFinal;

	sem_init(&enviePedido, 0, 0);

	//signal(SIGCHLD, sigchld_handler);

	switch (fork())
	{
	case -1 :

		perror("fork");
		break;

	case 0 : /*LLamada al proceso Consola PPD*/
		resultado = execv (configuracion.PathConsola, args);
		if (resultado == -1)
			perror("execv");
		cerrarArchivo();
		break;

	default : /*Codigo del padre, aca seguiria*/

		unlink(configuracion.SocketUNIX);
		sd = socket(AF_UNIX, SOCK_STREAM, 0);
		if (sd < 0)
		{
			perror("socket() failed");
			cerrarArchivo();
			break;
		}

		memset(&serveraddr, 0, sizeof(serveraddr));
		serveraddr.sun_family = AF_UNIX;
		strcpy(serveraddr.sun_path, configuracion.SocketUNIX);

		rc = bind(sd, (struct sockaddr *)&serveraddr, SUN_LEN(&serveraddr));
		if (rc < 0)
		{
			perror("bind() failed");
			cerrarArchivo();
			break;
		}

		rc = listen(sd, 10);
		if (rc< 0)
		{
			perror("listen() failed");
			cerrarArchivo();
			break;
		}

		sd = accept(sd, NULL, NULL);
		if (sd < 0)
		{
			perror("accept() failed");
			cerrarArchivo();
			break;
		}

		while(1)
		{
			rc = recv(sd, &cabeceraRecibo, sizeof(cabeceraRecibo), MSG_WAITALL);
			if (rc < 0)
			{
				perror("recv() failed");
				cerrarArchivo();
				break;
			}

			switch(cabeceraRecibo.type)
			{

			case TEMPO:

				rc = recv(sd, &pedidoTrace1, cabeceraRecibo.payloadlength, MSG_WAITALL);

				//UTILIZO EL MISMO STRUCT DEL PEDIDO TRACE 1, YA QUE ES UN COMANDO Y UN DIGITO
				configuracion.TiempoLectura = 1000000*pedidoTrace1.s1;

				break;

			case SIZEPEDIDOS:

					respuestaINFO = malloc(sizeof(protocoloINFO_respuesta));
					respuestaINFO->cabecera.type = SIZEPEDIDOS;
					respuestaINFO->cabecera.payloadlength = sizeof(int32_t);
					respuestaINFO->posicionActual = sizeCola;

				streamEnvio = RespuestaINFO_serializer(respuestaINFO);

				rc = send(sd, streamEnvio->data, streamEnvio->length, 0);

				free(respuestaINFO);
				free(streamEnvio);

				break;

				case INFOCONSOLA:

					respuestaINFO = malloc(sizeof(protocoloINFO_respuesta));
					respuestaINFO->cabecera.type = INFO;
					respuestaINFO->cabecera.payloadlength = sizeof(int32_t);
					respuestaINFO->posicionActual = ppd_CHStoLBA(posicionActualCabezal);

					streamEnvio = RespuestaINFO_serializer(respuestaINFO);

					rc = send(sd, streamEnvio->data, streamEnvio->length, 0);

					free(respuestaINFO);
					free(streamEnvio);
					break;

				case CLEAN:

					rc = recv(sd, &pedidoClean, cabeceraRecibo.payloadlength, MSG_WAITALL);

					//armo tantos pedidos como diferencia haya entre los sectores que recibo
					sectorInicial = ppd_LBAToCHS(pedidoClean.sectorInicial);
					sectorFinal = ppd_LBAToCHS(pedidoClean.sectorFinal);

					cantidadSectores = diferenciaEntreSectores(sectorInicial, sectorFinal);

					for (i = 0; i <= cantidadSectores; i++)
					{
						pedidoAPlanificar = armoPedidoAPlanificar(pedidoClean.sectorInicial+i, 'E');
						PPD_CHS estructura;
						estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);


							pthread_mutex_lock(&mutexListaPedidos);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock(&mutexListaPedidos);

					}
					break;

				case TRACE1:

					rc = recv(sd, &pedidoTrace1, cabeceraRecibo.payloadlength, MSG_WAITALL);

					sectorPlanifico = ppd_LBAToCHS(pedidoTrace1.s1);

					pedidoAPlanificar = armoPedidoAPlanificar(pedidoTrace1.s1, 'L');
					PPD_CHS estructura;
					estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

					pthread_mutex_lock(&mutexListaPedidos);
					agregarPedido_cola(pedidoAPlanificar);
					if (configuracion.LogActivado==1)
						log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
					pthread_mutex_unlock(&mutexListaPedidos);

					break;

					case TRACE2:

						rc = recv(sd, &pedidoTrace2, cabeceraRecibo.payloadlength, MSG_WAITALL);

						contador = 0;
						current = 0;
						sectoresAPlanificar = malloc(sizeof(int32_t)*2);
						memcpy(sectoresAPlanificar+current,&pedidoTrace2.s1, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace2.s2, sizeof(int32_t));


						while (contador < 2)
						{
							sectorPlanifico = ppd_LBAToCHS(sectoresAPlanificar[contador]);

							pedidoAPlanificar = armoPedidoAPlanificar(ppd_CHStoLBA(sectorPlanifico),'L');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

								pthread_mutex_lock(&mutexListaPedidos);
								agregarPedido_cola(pedidoAPlanificar);
								if (configuracion.LogActivado==1)
									log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
								pthread_mutex_unlock(&mutexListaPedidos);

							contador++;
						}

						free(sectoresAPlanificar);
						break;

					case TRACE3:

						rc = recv(sd, &pedidoTrace3, cabeceraRecibo.payloadlength, MSG_WAITALL);

						contador = 0;
						current = 0;
						sectoresAPlanificar = malloc(sizeof(int32_t)*3);
						memcpy(sectoresAPlanificar+current,&pedidoTrace3.s1, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace3.s2, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace3.s3, sizeof(int32_t));

						while(contador<3)
						{
							sectorPlanifico = ppd_LBAToCHS(sectoresAPlanificar[contador]);

							pedidoAPlanificar = armoPedidoAPlanificar(ppd_CHStoLBA(sectorPlanifico),'L');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

							pthread_mutex_lock(&mutexListaPedidos);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock(&mutexListaPedidos);

							contador++;
						}

						free(sectoresAPlanificar);
						break;

					case TRACE4:

						rc = recv(sd, &pedidoTrace4, cabeceraRecibo.payloadlength, MSG_WAITALL);

						contador = 0;
						current = 0;
						sectoresAPlanificar = malloc(sizeof(int32_t)*4);
						memcpy(sectoresAPlanificar+current,&pedidoTrace4.s1, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace4.s2, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace4.s3, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace4.s4, sizeof(int32_t));

						while(contador < 4)
						{
							sectorPlanifico = ppd_LBAToCHS(sectoresAPlanificar[contador]);

							pedidoAPlanificar = armoPedidoAPlanificar(ppd_CHStoLBA(sectorPlanifico),'L');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

							pthread_mutex_lock(&mutexListaPedidos);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock(&mutexListaPedidos);

							contador++;
						}

						free(sectoresAPlanificar);
						break;

					case TRACE5:

						rc = recv(sd, &pedidoTrace5, cabeceraRecibo.payloadlength, MSG_WAITALL);

						contador = 0;
						current = 0;
						sectoresAPlanificar = malloc(sizeof(int32_t)*5);
						memcpy(sectoresAPlanificar+current,&pedidoTrace5.s1, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace5.s2, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace5.s3, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace5.s4, sizeof(int32_t));
						current++;
						memcpy(sectoresAPlanificar+current, &pedidoTrace5.s5, sizeof(int32_t));

						while(contador < 5)
						{
							sectorPlanifico = ppd_LBAToCHS(sectoresAPlanificar[contador]);

							pedidoAPlanificar = armoPedidoAPlanificar(ppd_CHStoLBA(sectorPlanifico),'L');
							PPD_CHS estructura;
							estructura = ppd_LBAToCHS(pedidoAPlanificar.direccionSector);

							pthread_mutex_lock(&mutexListaPedidos);
							agregarPedido_cola(pedidoAPlanificar);
							if (configuracion.LogActivado==1)
								log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", pedidoAPlanificar.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_unlock(&mutexListaPedidos);

							contador++;
						}

						free(sectoresAPlanificar);
						break;
			}

			if (rc < 0)
			{
				perror("send() failed");
				cerrarArchivo();
				break;
			}
		}
		unlink(configuracion.SocketUNIX);
		close(sd);
		break;
	}
}

//void thread_lector_pedidos_pfs (void *parametro)
//{
//	struct_thread_lector struct_thread = *(struct_thread_lector *) parametro;
//
//	Nodo *lista, **aux = struct_thread.lista;
//	nipc paquete_recibo, paquete_mando;
//	int32_t addrlen = sizeof(struct sockaddr_in);
//	int32_t socket_server_pfs = socket(AF_INET,SOCK_STREAM,0);
//	int32_t cliente_pfs, rc;
//	fd_set sockets_pfs, sockets_select ;
//
//	int32_t clientes_pfs[configuracion.cantClientesMaxPFS];
//	int32_t descriptor_maximo = 0,a;
//
//	PPD_CHS estructura;
//
//	inicializoArray(clientes_pfs);
//	lista = *aux;
//
//	//Direccion local pfs
//	struct sockaddr_in *direccion_local_pfs = malloc(addrlen);
//	//Direccion remota(a la que me quiero conectar)
//	struct sockaddr_in *direccion_remota_pfs = malloc(addrlen);
//
//	direccion_local_pfs->sin_family = AF_INET;
//	direccion_local_pfs->sin_addr.s_addr = inet_addr("127.0.0.1");
//
//	direccion_local_pfs->sin_port = htons(configuracion.PuertoEscuchaPpd);
//	bind(socket_server_pfs,(struct sockaddr *)direccion_local_pfs, addrlen);
//
//	//loguear
//
//	rc = listen(socket_server_pfs,configuracion.cantClientesMaxPFS);
//	if (rc < 0)
//	{
//		if (configuracion.LogActivado==1)
//			log_error(logger, "THREAD_PFS", "El PPD ha esperado demasiado por conexiones entrantes");
//		cerrarArchivo();
//		perror("listen() failed");
//	}
//
//	//loguear
//
//	FD_ZERO (&sockets_pfs);
//	FD_SET(socket_server_pfs,&sockets_pfs);
//	descriptor_maximo = socket_server_pfs;
//
//	while(1)
//	{
//		sockets_select = sockets_pfs;
//
//		select(descriptor_maximo+1,&sockets_select,NULL,NULL,NULL);
//
//		if(FD_ISSET(socket_server_pfs,&sockets_select))
//		{
//			rc = accept(socket_server_pfs,(struct sockaddr *)direccion_remota_pfs,(void *)&addrlen);
//			cliente_pfs = rc;
//			if (configuracion.LogActivado==1)
//				log_info(logger, "THREAD_PFS", "Se acepto la conexion del CLIENTE PFS %i \n", cliente_pfs);
//
//
//			insertarEnArray(clientes_pfs,cliente_pfs);
//			FD_SET(cliente_pfs,&sockets_pfs);
//			if(cliente_pfs > descriptor_maximo)
//				descriptor_maximo = cliente_pfs;
//
//		}
//
//		for(a=0; a<configuracion.cantClientesMaxPFS; a++)
//		{
//			if(FD_ISSET (clientes_pfs[a],&sockets_select))
//			{
//				rc = recv(clientes_pfs[a],&paquete_recibo,sizeof(nipc),0);
//				switch(rc)
//				{
//				case 0: //EL CLIENTE CERRO LA CONEXION
//					eliminoDeArray(clientes_pfs,clientes_pfs[a]);
//					close(clientes_pfs[a]);
//					if (configuracion.LogActivado==1)
//						log_error(logger, "THREAD_PFS", "El cliente %i cerro la conexion \ņ", clientes_pfs[a]);
//					break;
//				case -1: //ERROR EN EL RECV
//					//LOGUEO ERROR
//					if (configuracion.LogActivado==1)
//						log_error(logger, "THREAD_PFS", "Error al recibir el paquete, se cierra la conexion del cliente %i \ņ", clientes_pfs[a]);
//					close(clientes_pfs[a]);
//					break;
//				default: //SE RECIBIO CORRECTAMENTE
//					switch(paquete_recibo.type)
//					{
//					case HANDSHAKE:
//						if(paquete_recibo.payloadlength == 0)
//						{
//							paquete_mando.type = HANDSHAKE;
//							paquete_mando.payloadlength = 0;
//							paquete_mando.payload.conexion_pfs_id = clientes_pfs[a];
//							if (configuracion.LogActivado==1)
//								log_info(logger, "THREAD_PFS", "Se recibio el HANDSHAKE correctamente \ņ");
//							send(clientes_pfs[a],&paquete_mando,sizeof(nipc),0);
//						}
//						else
//						{
//							paquete_mando.type = HANDSHAKE; //HANDSHAKE INCORRECTO
//							paquete_mando.payloadlength = sizeof(nipc);
//							strcpy(paquete_mando.payload.info,"HANDSHAKE INCORRECTO");
//							if (configuracion.LogActivado==1)
//								log_info(logger, "THREAD_PFS", "Se recibio un HANDSHAKE incorrecto \ņ");
//							send(clientes_pfs[a],&paquete_mando,sizeof(nipc),0);
//						}
//						break;
//					default:
//						//EL PEDIDO QUE LLEGO NO ES DE TIPO HANDSHAKE
//						switch (algoritmoUtilizado)
//						{
//						case CON_INANICION:
//							estructura = ppd_LBAToCHS(paquete_recibo.payload.direccionSector);
//							if (configuracion.LogActivado==1)
//								log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", paquete_recibo.payload.lectura_o_escritura, estructura.cylinder, estructura.sector);
//							pthread_mutex_lock (&mutexColaSSTF);
//							agregarPedido_cola(paquete_recibo.payload);
//							pthread_mutex_unlock (&mutexColaSSTF);
//							break;
//						case SIN_INANICION:
//							estructura = ppd_LBAToCHS(paquete_recibo.payload.direccionSector);
//							if (configuracion.LogActivado==1)
//								log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", paquete_recibo.payload.lectura_o_escritura, estructura.cylinder, estructura.sector);
//							pthread_mutex_lock (&mutexListaPedidos);
//							agregaPedidoEnListaNstep(struct_thread.listaNstep,paquete_recibo.payload);
//							pthread_mutex_unlock (&mutexListaPedidos);
//							break;
//						}
//						break;
//					}
//					break;
//				}
//			}
//		}
//	}
//}
void thread_lector_pedidos_pfs (void *parametro)
{
	struct_thread_lector struct_thread = *(struct_thread_lector *) parametro;

	Nodo *lista, **aux = struct_thread.lista;
	nipc paquete_recibo, paquete_mando;
	int32_t addrlen = sizeof(struct sockaddr_in);
	int32_t socket_server_pfs = socket(AF_INET,SOCK_STREAM,0);
	int32_t cliente_pfs, rc;
	fd_set sockets_pfs, sockets_select ;

	int32_t clientes_pfs[configuracion.cantClientesMaxPFS];
	int32_t descriptor_maximo = 0,a;

	PPD_CHS estructura;

	inicializoArray(clientes_pfs);
	lista = *aux;

	//Direccion local pfs
	struct sockaddr_in *direccion_local_pfs = malloc(addrlen);
	//Direccion remota(a la que me quiero conectar)
	struct sockaddr_in *direccion_remota_pfs = malloc(addrlen);

	direccion_local_pfs->sin_family = AF_INET;
	direccion_local_pfs->sin_addr.s_addr = inet_addr("127.0.0.1");

	direccion_local_pfs->sin_port = htons(configuracion.PuertoEscuchaPpd);
	bind(socket_server_pfs,(struct sockaddr *)direccion_local_pfs, addrlen);

	//loguear

	rc = listen(socket_server_pfs,CANT_CLIENTES_PFS_MAX);
	if (rc < 0)
	{
		if (configuracion.LogActivado==1)
			log_error(logger, "THREAD_PFS", "El PPD ha esperado demasiado por conexiones entrantes");
		cerrarArchivo();
		perror("listen() failed");
	}

	//loguear

	FD_ZERO (&sockets_pfs);
	FD_SET(socket_server_pfs,&sockets_pfs);
	descriptor_maximo = socket_server_pfs;

	//

	while(1)
	{
		sockets_select = sockets_pfs;

		select(descriptor_maximo+1,&sockets_select,NULL,NULL,NULL);

		if(FD_ISSET(socket_server_pfs,&sockets_select))
		{
			rc = accept(socket_server_pfs,(struct sockaddr *)direccion_remota_pfs,(void *)&addrlen);
			cliente_pfs = rc;
			if (configuracion.LogActivado==1)
				log_info(logger, "THREAD_PFS", "Se acepto la conexion del CLIENTE PFS %i", cliente_pfs);

			//Inserto en la lista
			insertarEnListaClientes(cliente_pfs);
			FD_SET(cliente_pfs,&sockets_pfs);
			if(cliente_pfs > descriptor_maximo)
				descriptor_maximo = cliente_pfs;

		}

		//NodoPFS *listaPFS=NULL;
		NodoPFS *auxPFS=NULL;

		auxPFS=listaPFS;

		while(auxPFS!=NULL){
			if(FD_ISSET (auxPFS->cliente_pfs,&sockets_select))
			{
				rc = recv(auxPFS->cliente_pfs,&paquete_recibo,sizeof(nipc),0);
				switch(rc)
				{
				case 0: //EL CLIENTE CERRO LA CONEXION
					borrarDeListaClientes(auxPFS->cliente_pfs);
					close(auxPFS->cliente_pfs);
					if (configuracion.LogActivado==1)
						log_error(logger, "THREAD_PFS", "El cliente %i cerro la conexion", auxPFS->cliente_pfs);
					break;
				case -1: //ERROR EN EL RECV
					//LOGUEO ERROR
					if (configuracion.LogActivado==1)
						log_error(logger, "THREAD_PFS", "Error al recibir el paquete, se cierra la conexion del cliente %i", auxPFS->cliente_pfs);
					close(auxPFS->cliente_pfs);
					break;
				default: //SE RECIBIO CORRECTAMENTE
					switch(paquete_recibo.type)
					{
					case HANDSHAKE:
						if(paquete_recibo.payloadlength == 0)
						{
							paquete_mando.type = HANDSHAKE;
							paquete_mando.payloadlength = 0;
							paquete_mando.payload.conexion_pfs_id = auxPFS->cliente_pfs;
							if (configuracion.LogActivado==1)
								log_info(logger, "THREAD_PFS", "Se recibio el HANDSHAKE correctamente");
							send(auxPFS->cliente_pfs,&paquete_mando,sizeof(nipc),0);
						}
						else
						{
							paquete_mando.type = HANDSHAKE; //HANDSHAKE INCORRECTO
							paquete_mando.payloadlength = sizeof(nipc);
							strcpy(paquete_mando.payload.info,"HANDSHAKE INCORRECTO");
							if (configuracion.LogActivado==1)
								log_info(logger, "THREAD_PFS", "Se recibio un HANDSHAKE incorrecto");
							send(auxPFS->cliente_pfs,&paquete_mando,sizeof(nipc),0);
						}
						//break;
						//EL PEDIDO QUE LLEGO NO ES DE TIPO HANDSHAKE
							estructura = ppd_LBAToCHS(paquete_recibo.payload.direccionSector);
							if (configuracion.LogActivado==1)
								log_info(logger, "MAIN", "Se agrego un pedido de %c sobre el sector %d:%d a la cola \n", paquete_recibo.payload.lectura_o_escritura, estructura.cylinder, estructura.sector);
							pthread_mutex_lock (&mutexListaPedidos);
							agregarPedido_cola(paquete_recibo.payload);
							pthread_mutex_unlock (&mutexListaPedidos);

					}
					break;
				}
			}
			auxPFS=auxPFS->sig;
		}
		//Aca finaliza la carga eso
	}
}



void insertarEnListaClientes(int32_t cliente_pfs){
	NodoPFS *aux;

	// creo el nodo a insertar
	NodoPFS *nuevo=(NodoPFS*)malloc(sizeof(NodoPFS));

	nuevo->sig=NULL;
	nuevo->cliente_pfs=cliente_pfs;

	if( listaPFS == NULL )
	{
		listaPFS = nuevo;
	}
	else
	{
		aux = listaPFS;
		while( aux->sig != NULL )
		{
			aux = aux->sig;
		}

		aux->sig = nuevo;
	}

	cantidadClientesPFS++;
}

void borrarDeListaClientes(int32_t cliente_pfs){

	NodoPFS *aux, *anterior;

	anterior = NULL;

	for (aux = listaPFS;
			aux != NULL;
			anterior = aux, aux = aux->sig) {


		if (((aux->cliente_pfs == cliente_pfs))) {  /* Encontrado */
			if (anterior == NULL) {

				listaPFS = aux->sig;
			} else {

				anterior->sig = aux->sig;
			}

			free(aux);

			// cantidadTotalPedidos--;

			return;
		}
	}
	cantidadClientesPFS--;

}



